package com.eshipenet.shipnet.edgeboxextbuilder
import java.lang

import org.apache.flink.api.common.functions.CoGroupFunction
import org.apache.flink.util.Collector

object EdgeBoxModelChangeFunction {

  type T = (String, Set[EdgeBox])

}

import EdgeBoxModelChangeFunction._

class EdgeBoxModelChangeFunction extends CoGroupFunction[T, EdgeBoxModelAction, EdgeBoxExtAction] {
  override def coGroup(first: lang.Iterable[T],
                       second: lang.Iterable[EdgeBoxModelAction],
                       out: Collector[EdgeBoxExtAction]): Unit = {
    println("first: ", first)
    println("second: ", second)
    second.forEach {
      case EdgeBoxModelInsert(id, after) =>
        first.forEach { kv =>
          if (kv._1 == id) {
            kv._2.foreach { edgeBox =>
              out.collect(EdgeBoxExtModelUpdate(edgeBox.id, edgeBox, Some(after)))
            }
          }
        }
      case EdgeBoxModelUpdate(id, before, after) =>
        first.forEach { kv =>
          if (kv._1 == id) {
            kv._2.foreach { edgeBox =>
              out.collect(EdgeBoxExtModelUpdate(edgeBox.id, edgeBox, Some(after)))
            }
          }
        }
      case EdgeBoxModelDelete(id, before) =>
        first.forEach { kv =>
          if (kv._1 == id) {
            kv._2.foreach { edgeBox =>
              out.collect(EdgeBoxExtModelUpdate(edgeBox.id, edgeBox, None))
            }
          }
        }
    }
  }
}
